package com.wande.dataplatform.filecollection.scheduler;

import cn.hutool.json.JSONUtil;
import com.wande.dataplatform.filecollection.common.constant.FileCollectionConstants;
import com.wande.dataplatform.filecollection.common.enums.FileCollectionErrorCode;
import com.wande.dataplatform.filecollection.common.exception.FileCollectionException;
import com.wande.dataplatform.filecollection.domain.CollectionRule;
import com.wande.dataplatform.filecollection.executor.CollectionTaskExecutor;
import com.wande.dataplatform.filecollection.mapper.CollectionRuleMapper;
import com.wande.dataplatform.service.IDolphinSchedulerService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

/**
 * 文件采集调度器
 *
 * @author wande
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class FileCollectionScheduler {

    private final IDolphinSchedulerService dolphinSchedulerService;
    private final CollectionRuleMapper collectionRuleMapper;
    private final CollectionTaskExecutor collectionTaskExecutor;

    /**
     * 创建调度任务
     *
     * @param rule 采集规则
     * @return DolphinScheduler任务ID
     */
    public Long createScheduleTask(CollectionRule rule) {
        try {
            // 如果是手动触发，不创建调度任务
            if (FileCollectionConstants.SCHEDULE_TYPE_MANUAL.equals(rule.getScheduleType())) {
                return null;
            }

            // 构建工作流定义
            Map<String, Object> workflowDef = buildWorkflowDefinition(rule);
            
            // 调用DolphinScheduler API创建工作流
            // 注意：这里需要根据实际的DolphinScheduler API进行调整
            Long dsTaskId = createDolphinSchedulerTask(workflowDef);
            
            log.info("创建调度任务成功: ruleId={}, dsTaskId={}", rule.getId(), dsTaskId);
            return dsTaskId;
            
        } catch (Exception e) {
            log.error("创建调度任务失败: ruleId={}", rule.getId(), e);
            throw new FileCollectionException(FileCollectionErrorCode.SCHEDULE_CREATE_ERROR,
                "创建调度任务失败: " + e.getMessage());
        }
    }

    /**
     * 更新调度任务
     *
     * @param rule 采集规则
     */
    public void updateScheduleTask(CollectionRule rule) {
        try {
            if (rule.getDsTaskId() == null) {
                return;
            }

            // 如果改为手动触发，删除调度任务
            if (FileCollectionConstants.SCHEDULE_TYPE_MANUAL.equals(rule.getScheduleType())) {
                deleteScheduleTask(rule.getDsTaskId());
                rule.setDsTaskId(null);
                collectionRuleMapper.updateById(rule);
                return;
            }

            // 构建工作流定义
            Map<String, Object> workflowDef = buildWorkflowDefinition(rule);
            
            // 调用DolphinScheduler API更新工作流
            updateDolphinSchedulerTask(rule.getDsTaskId(), workflowDef);
            
            log.info("更新调度任务成功: ruleId={}, dsTaskId={}", rule.getId(), rule.getDsTaskId());
            
        } catch (Exception e) {
            log.error("更新调度任务失败: ruleId={}", rule.getId(), e);
            throw new FileCollectionException(FileCollectionErrorCode.SCHEDULE_UPDATE_ERROR,
                "更新调度任务失败: " + e.getMessage());
        }
    }

    /**
     * 删除调度任务
     *
     * @param dsTaskId DolphinScheduler任务ID
     */
    public void deleteScheduleTask(Long dsTaskId) {
        try {
            if (dsTaskId == null) {
                return;
            }

            // 调用DolphinScheduler API删除工作流
            dolphinSchedulerService.deleteWorkflow(dsTaskId);
            
            log.info("删除调度任务成功: dsTaskId={}", dsTaskId);
            
        } catch (Exception e) {
            log.error("删除调度任务失败: dsTaskId={}", dsTaskId, e);
            throw new FileCollectionException(FileCollectionErrorCode.SCHEDULE_DELETE_ERROR,
                "删除调度任务失败: " + e.getMessage());
        }
    }

    /**
     * 手动触发任务
     *
     * @param ruleId 规则ID
     * @return 任务ID
     */
    public Long triggerTask(Long ruleId) {
        try {
            CollectionRule rule = collectionRuleMapper.selectById(ruleId);
            if (rule == null) {
                throw new FileCollectionException(FileCollectionErrorCode.RULE_NOT_FOUND,
                    "采集规则不存在: " + ruleId);
            }

            // 检查规则状态
            if (!FileCollectionConstants.RULE_STATUS_ENABLED.equals(rule.getStatus())) {
                throw new FileCollectionException(FileCollectionErrorCode.RULE_DISABLED,
                    "采集规则未启用");
            }

            // 执行采集任务
            Long taskId = collectionTaskExecutor.executeTask(rule);
            
            log.info("手动触发任务成功: ruleId={}, taskId={}", ruleId, taskId);
            return taskId;
            
        } catch (Exception e) {
            log.error("手动触发任务失败: ruleId={}", ruleId, e);
            throw new FileCollectionException(FileCollectionErrorCode.TASK_TRIGGER_ERROR,
                "手动触发任务失败: " + e.getMessage());
        }
    }

    /**
     * 暂停任务
     *
     * @param ruleId 规则ID
     */
    public void pauseTask(Long ruleId) {
        try {
            CollectionRule rule = collectionRuleMapper.selectById(ruleId);
            if (rule == null || rule.getDsTaskId() == null) {
                return;
            }

            // 调用DolphinScheduler API暂停工作流
            // 注意：这里需要根据实际的DolphinScheduler API进行调整
            pauseDolphinSchedulerTask(rule.getDsTaskId());
            
            log.info("暂停任务成功: ruleId={}, dsTaskId={}", ruleId, rule.getDsTaskId());
            
        } catch (Exception e) {
            log.error("暂停任务失败: ruleId={}", ruleId, e);
            throw new FileCollectionException(FileCollectionErrorCode.TASK_PAUSE_ERROR,
                "暂停任务失败: " + e.getMessage());
        }
    }

    /**
     * 恢复任务
     *
     * @param ruleId 规则ID
     */
    public void resumeTask(Long ruleId) {
        try {
            CollectionRule rule = collectionRuleMapper.selectById(ruleId);
            if (rule == null || rule.getDsTaskId() == null) {
                return;
            }

            // 调用DolphinScheduler API恢复工作流
            // 注意：这里需要根据实际的DolphinScheduler API进行调整
            resumeDolphinSchedulerTask(rule.getDsTaskId());
            
            log.info("恢复任务成功: ruleId={}, dsTaskId={}", ruleId, rule.getDsTaskId());
            
        } catch (Exception e) {
            log.error("恢复任务失败: ruleId={}", ruleId, e);
            throw new FileCollectionException(FileCollectionErrorCode.TASK_RESUME_ERROR,
                "恢复任务失败: " + e.getMessage());
        }
    }

    /**
     * 任务状态回调处理
     *
     * @param dsTaskId DolphinScheduler任务ID
     * @param status 任务状态
     */
    public void handleTaskCallback(Long dsTaskId, String status) {
        try {
            log.info("收到任务状态回调: dsTaskId={}, status={}", dsTaskId, status);
            
            // 根据状态执行相应的处理逻辑
            // 这里可以更新任务状态、发送通知等
            
        } catch (Exception e) {
            log.error("处理任务回调失败: dsTaskId={}", dsTaskId, e);
        }
    }

    /**
     * 构建工作流定义
     */
    private Map<String, Object> buildWorkflowDefinition(CollectionRule rule) {
        Map<String, Object> workflowDef = new HashMap<>();
        
        // 工作流基本信息
        workflowDef.put("name", "file_collection_" + rule.getRuleCode());
        workflowDef.put("description", rule.getRuleName());
        
        // 调度配置
        Map<String, Object> schedule = new HashMap<>();
        schedule.put("cron", rule.getScheduleConfig());
        schedule.put("startTime", null);
        schedule.put("endTime", null);
        workflowDef.put("schedule", schedule);
        
        // 任务定义
        Map<String, Object> taskDef = new HashMap<>();
        taskDef.put("type", "SHELL");
        taskDef.put("name", "file_collection_task");
        
        // Shell脚本：调用采集任务执行器
        String script = String.format(
            "curl -X POST http://localhost:8080/api/dataplatform/file-collection/trigger/%d",
            rule.getId()
        );
        taskDef.put("script", script);
        
        workflowDef.put("tasks", new Object[]{taskDef});
        
        return workflowDef;
    }

    /**
     * 创建DolphinScheduler任务
     */
    private Long createDolphinSchedulerTask(Map<String, Object> workflowDef) {
        // 这里需要根据实际的DolphinScheduler API进行实现
        // 示例代码：
        log.info("创建DolphinScheduler任务: {}", JSONUtil.toJsonStr(workflowDef));
        
        // 返回模拟的任务ID
        return System.currentTimeMillis();
    }

    /**
     * 更新DolphinScheduler任务
     */
    private void updateDolphinSchedulerTask(Long dsTaskId, Map<String, Object> workflowDef) {
        // 这里需要根据实际的DolphinScheduler API进行实现
        log.info("更新DolphinScheduler任务: dsTaskId={}, def={}", dsTaskId, JSONUtil.toJsonStr(workflowDef));
    }

    /**
     * 暂停DolphinScheduler任务
     */
    private void pauseDolphinSchedulerTask(Long dsTaskId) {
        // 这里需要根据实际的DolphinScheduler API进行实现
        log.info("暂停DolphinScheduler任务: dsTaskId={}", dsTaskId);
    }

    /**
     * 恢复DolphinScheduler任务
     */
    private void resumeDolphinSchedulerTask(Long dsTaskId) {
        // 这里需要根据实际的DolphinScheduler API进行实现
        log.info("恢复DolphinScheduler任务: dsTaskId={}", dsTaskId);
    }
}
